﻿using NetMQ;
using NetMQ.Sockets;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading;

namespace MicService.MS4Net.Demo.FindService
{
    /// <summary>
    /// 使用NetMQ中的NetMQBeacon做本地网络的端到端的服务发现
    /// </summary>
    public class FindService
    {
        private static FindService _instance;

        public static FindService Instance
        {
            get
            {
                if (_instance == null)
                    _instance = new FindService();
                return _instance;
            }
        }

        private FindService()
        {
        }

        private NetMQActor Actor { get; set; }
        private string HostAddress { get; set; }

        public void Init()
        {
            var actor = Bus.Create(9999);

            actor.SendFrame(Bus.GetHostAddressCommand);
            var hostAddress = actor.ReceiveFrameString();

            Console.Title = $"NetMQ Beacon Demo at {hostAddress}";

            Actor = actor;

            HostAddress = hostAddress;
        }

        public void SayHello()
        {
            // beacons publish every second, so wait a little longer than that to
            // let all the other nodes connect to our new node
            Thread.Sleep(1100);
            // publish a hello message
            // note we can use NetMQSocket send and receive extension methods
            Actor.SendMoreFrame(Bus.PublishCommand).SendMoreFrame("Hello?").SendFrame(HostAddress);
        }

        public void StartListenMsg()
        {
            if (MsgCache == null)
                MsgCache = new Queue<string>();
            if (mListener == null)
                mListener = new Thread(new ThreadStart(DoListen));
            if (mListener.ThreadState == ThreadState.Unstarted)
                mListener.Start();
            if (mStopListenREvent == null)
                mStopListenREvent = new ManualResetEvent(false);
            mStopListenREvent.Reset();
        }

        public void StopListenMsg()
        {
            mStopListenREvent.Set();
        }

        private ManualResetEvent mStopListenREvent;
        private Queue<string> MsgCache { get; set; }
        private Thread mListener { get; set; }

        private void ListenWork()
        {
            while (true)
            {
                if (mStopListenREvent.WaitOne())
                    break;
                DoListen();
            }
        }

        private void DoListen()
        {
            var actor = Actor;
            var hostAddress = HostAddress;
            string message = actor.ReceiveFrameString();
            MsgCache.Enqueue(message);
            switch (message)
            {
                case "Hello?":
                    // another node is saying hello
                    var fromHostAddress = actor.ReceiveFrameString();
                    var msg = fromHostAddress + " says Hello?";
                    Console.WriteLine(msg);

                    // send back a welcome message via the Bus publisher
                    msg = hostAddress + " says Welcome!";
                    actor.SendMoreFrame(Bus.PublishCommand).SendFrame(msg);
                    break;

                case Bus.AddedNodeCommand:
                    var addedAddress = actor.ReceiveFrameString();
                    Console.WriteLine("Added node {0} to the Bus", addedAddress);
                    break;

                case Bus.RemovedNodeCommand:
                    var removedAddress = actor.ReceiveFrameString();
                    Console.WriteLine("Removed node {0} from the Bus", removedAddress);
                    break;

                default:
                    // it's probably a welcome message
                    Console.WriteLine(message);
                    break;
            }
        }

        private static void Main_Demo()
        {
            Console.Title = "NetMQ Beacon Demo";

            // Create a bus using broadcast port 9999
            // All communication with the bus is through the returned actor
            var actor = Bus.Create(9999);

            actor.SendFrame(Bus.GetHostAddressCommand);
            var hostAddress = actor.ReceiveFrameString();

            Console.Title = $"NetMQ Beacon Demo at {hostAddress}";

            // beacons publish every second, so wait a little longer than that to
            // let all the other nodes connect to our new node
            Thread.Sleep(1100);

            // publish a hello message
            // note we can use NetMQSocket send and receive extension methods
            actor.SendMoreFrame(Bus.PublishCommand).SendMoreFrame("Hello?").SendFrame(hostAddress);

            // receive messages from other nodes on the bus
            while (true)
            {
                // actor is receiving messages forwarded by the Bus subscriber
                string message = actor.ReceiveFrameString();
                switch (message)
                {
                    case "Hello?":
                        // another node is saying hello
                        var fromHostAddress = actor.ReceiveFrameString();
                        var msg = fromHostAddress + " says Hello?";
                        Console.WriteLine(msg);

                        // send back a welcome message via the Bus publisher
                        msg = hostAddress + " says Welcome!";
                        actor.SendMoreFrame(Bus.PublishCommand).SendFrame(msg);
                        break;

                    case Bus.AddedNodeCommand:
                        var addedAddress = actor.ReceiveFrameString();
                        Console.WriteLine("Added node {0} to the Bus", addedAddress);
                        break;

                    case Bus.RemovedNodeCommand:
                        var removedAddress = actor.ReceiveFrameString();
                        Console.WriteLine("Removed node {0} from the Bus", removedAddress);
                        break;

                    default:
                        // it's probably a welcome message
                        Console.WriteLine(message);
                        break;
                }
            }
        }
    }

    /// <summary>
    /// 节点信息
    /// </summary>
    public class NodeKey
    {
        public NodeKey(string name, int port)
        {
            Name = name;
            Port = port;
            Address = $"tcp://{name}:{port}";
            HostName = Dns.GetHostEntry(name).HostName;
        }

        public string Name { get; }
        public int Port { get; }

        public string Address { get; }

        public string HostName { get; private set; }

        protected bool Equals(NodeKey other)
        {
            return string.Equals(Name, other.Name) && Port == other.Port;
        }

        public override bool Equals(object obj)
        {
            if (ReferenceEquals(null, obj))
                return false;
            if (ReferenceEquals(this, obj))
                return true;
            if (obj.GetType() != this.GetType())
                return false;
            return Equals((NodeKey)obj);
        }

        public override int GetHashCode()
        {
            unchecked
            {
                return ((Name?.GetHashCode() ?? 0) * 397) ^ Port;
            }
        }

        public override string ToString()
        {
            return Address;
        }
    }

    public class BusCmd
    {
        public const string CMD_Publish = "Publish";
        public const string CMD_Hello = "Hello";
        public const string CMD_Exit = "Exit";
        public const string CMD_ADDNODE = "AddNode";
        public const string CMD_DELNODE = "DelNode";
        public const string CMD_GETHOSTADDR = "GetHostAddr";
    }

    public class MyBus:IDisposable
    {
        private readonly NetMQActor m_actor;
        private PublisherSocket m_publisher;
        private SubscriberSocket m_subscriber;
        private NetMQBeacon m_beacon;
        private NetMQPoller m_poller;
        private PairSocket m_shim;
        private NetMQTimer m_mqtimer;

        private readonly TimeSpan m_deadNodeTimeout = TimeSpan.FromSeconds(10);
        private readonly Dictionary<NodeKey, DateTime> m_nodes;
        private readonly int m_broadcastPort;
        private int m_randomPort;
        private string mLastEndpoint;
        private bool disposedValue;

        public string HostAddress { get; private set; }

        public MyBus(int port)
        {
            m_nodes = new Dictionary<NodeKey, DateTime>();
            m_broadcastPort = port;
            m_actor = NetMQActor.Create(RunActor);
            InitHostAddress();
        }

        private void RunActor(PairSocket shim)
        {
            // save the shim to the class to use later
            m_shim = shim;

            // create all subscriber, publisher and beacon
            using (m_subscriber = new SubscriberSocket())
            using (m_publisher = new PublisherSocket())
            using (m_beacon = new NetMQBeacon())
            {
                //订阅所有的消息
                m_subscriber.Subscribe("");
                //订阅者绑定一个随机的端口，随后使用 beacon发布该端口
                m_randomPort = m_subscriber.BindRandomPort("tcp://*");
                mLastEndpoint = m_subscriber.Options.LastEndpoint;
                //给beacon配置特定的端口
                m_beacon.Configure(m_broadcastPort);
                //使用beacon将随机端口发布到所有的其他节点
                m_beacon.Publish(m_randomPort.ToString(), TimeSpan.FromSeconds(1));// publishing the random port to all other nodes
                //beacon订阅所有的其他已配置特定端口的beacon的所有的UDP消息
                m_beacon.Subscribe("");
                //创建一个用于清理下线的结点的定时器
                NetMQTimer timer = new NetMQTimer(TimeSpan.FromSeconds(1));
                m_mqtimer = timer;
                //创建一个轮询器
                m_poller = new NetMQPoller();
                m_poller.Add(m_shim);
                m_poller.Add(m_subscriber);
                m_poller.Add(m_beacon);
                m_poller.Add(timer);
                //监听所有的actor的命令
                m_shim.ReceiveReady += OnShimReady;
                //监听其他发布者发送的消息，并推送给shim
                m_subscriber.ReceiveReady += OnSubscriberReady;
                //监听所有新加入的beacon
                m_beacon.ReceiveReady += OnBeaconReady;
                //定时器时间到执行下线节点的清理
                timer.Elapsed += ClearDeadNodes;
                //告诉actor已完成配置开始执行
                m_shim.SignalOK();
                //开始轮询直到取消
                m_poller.Run();
            }
        }

        #region 服务发现相关事件

        /// <summary>
        /// 清理下线节点
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void ClearDeadNodes(object sender, NetMQTimerEventArgs e)
        {
            var deadNodes = m_nodes
                .Where(n => DateTime.Now > n.Value + m_deadNodeTimeout)
                .Select(n => n.Key).ToArray();
            foreach (var node in deadNodes)
            {
                m_nodes.Remove(node);
                m_publisher.Disconnect(node.Address);
                m_shim.SendMoreFrame(BusCmd.CMD_DELNODE).SendFrame(node.Address);
            }
        }

        /// <summary>
        /// 有新节点加入
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnBeaconReady(object sender, NetMQBeaconEventArgs e)
        {
            var message = m_beacon.Receive();
            int port;
            int.TryParse(message.String, out port);
            NodeKey node = new NodeKey(message.PeerHost, port);
            if (!m_nodes.ContainsKey(node))
            {
                //缓存新节，连接新节点
                m_nodes.Add(node, DateTime.Now);
                m_publisher.Connect(node.Address);
                //发布新加入节点消息，公布新节点地址
                m_shim.SendMoreFrame(BusCmd.CMD_ADDNODE).SendFrame(node.Address);
            }
            else
            {
                //更新已有节点的存活时间
                m_nodes[node] = DateTime.Now;
            }
        }

        /// <summary>
        /// 监听其他发布者发送的消息，并推送给shim
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnSubscriberReady(object sender, NetMQSocketEventArgs e)
        {
            NetMQMessage message = m_subscriber.ReceiveMultipartMessage();
            m_shim.SendMultipartMessage(message);
        }

        /// <summary>
        /// 监听所有的actor的命令
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnShimReady(object sender, NetMQSocketEventArgs e)
        {
            string command = m_shim.ReceiveFrameString();
            if (command == NetMQActor.EndShimMessage)
            {
                // we cancel the socket which dispose and exist the shim
                m_poller.Stop();
            }
            else if (command == BusCmd.CMD_Publish)
            {
                // it is a publish command we just forward everything to the publisher until end of message
                NetMQMessage message = m_shim.ReceiveMultipartMessage();
                m_publisher.SendMultipartMessage(message);
            }
            else if (command == BusCmd.CMD_GETHOSTADDR)
            {
                var address = m_beacon.BoundTo + ":" + m_randomPort;
                m_shim.SendFrame(address);
            }
        }

        #endregion 服务发现相关事件

        private void InitHostAddress()
        {
            m_actor.SendFrame(Bus.GetHostAddressCommand);
            var hostAddress = m_actor.ReceiveFrameString();
            Console.Title = $"NetMQ Beacon Demo at {hostAddress}";
            HostAddress = hostAddress;
        }

        public void Hello(string msg)
        {
            m_actor.SendMoreFrame(Bus.PublishCommand).SendMoreFrame(msg).SendFrame(HostAddress);
        }

        public void Publish(string msg)
        {
            m_actor.SendMoreFrame(Bus.PublishCommand).SendMoreFrame(msg).SendFrame(HostAddress);
        }

        public string RecvMsg()
        {
            return m_actor?.ReceiveFrameString();
        }

        public void Stop()
        {
            m_actor.SendMoreFrame(NetMQActor.EndShimMessage);
        }

        public void Start()
        {
            m_actor.SignalOK();
            m_poller.Run();
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    // TODO: 释放托管状态(托管对象)
                    m_poller.Stop();
                    m_actor.Dispose();
                    m_publisher.Dispose();
                    m_beacon.Dispose();
                    m_shim.Dispose();
                    m_subscriber.Dispose();
                    m_mqtimer.Enable = false;
                }
                // TODO: 释放未托管的资源(未托管的对象)并重写终结器
                // TODO: 将大型字段设置为 null
                disposedValue = true;
            }
        }

        // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
        // ~MyBus()
        // {
        //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        //     Dispose(disposing: false);
        // }

        public void Dispose()
        {
            // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
    }

    /// <summary>
    /// Originally from the Docs\beacon.md documentation with small modifications for this demo
    /// </summary>
    internal class Bus
    {
        // Actor Protocol
        public const string PublishCommand = "P";

        public const string GetHostAddressCommand = "GetHostAddress";
        public const string AddedNodeCommand = "AddedNode";
        public const string RemovedNodeCommand = "RemovedNode";

        // Dead nodes timeout
        private readonly TimeSpan m_deadNodeTimeout = TimeSpan.FromSeconds(10);

        /// <summary>
        /// 广播端口
        /// </summary>
        private readonly int m_broadcastPort;

        private readonly NetMQActor m_actor;
        private PublisherSocket m_publisher;
        private SubscriberSocket m_subscriber;
        private NetMQBeacon m_beacon;
        private NetMQPoller m_poller;
        private PairSocket m_shim;
        private readonly Dictionary<NodeKey, DateTime> m_nodes; // value is the last time we "saw" this node
        private int m_randomPort;

        private Bus(int broadcastPort)
        {
            m_nodes = new Dictionary<NodeKey, DateTime>();
            m_broadcastPort = broadcastPort;
            m_actor = NetMQActor.Create(RunActor);
        }

        /// <summary>
        /// Creates a new message bus actor. All communication with the bus is
        /// through the returned <see cref="NetMQActor"/>.
        /// </summary>
        public static NetMQActor Create(int broadcastPort)
        {
            Bus node = new Bus(broadcastPort);
            return node.m_actor;
        }

        private void RunActor(PairSocket shim)
        {
            // save the shim to the class to use later
            m_shim = shim;

            // create all subscriber, publisher and beacon
            using (m_subscriber = new SubscriberSocket())
            using (m_publisher = new PublisherSocket())
            using (m_beacon = new NetMQBeacon())
            {
                // listen to actor commands
                m_shim.ReceiveReady += OnShimReady;

                // subscribe to all messages
                m_subscriber.Subscribe("");

                // we bind to a random port, we will later publish this port
                // using the beacon
                m_randomPort = m_subscriber.BindRandomPort("tcp://*");
                Console.WriteLine("Bus subscriber is bound to {0}", m_subscriber.Options.LastEndpoint);

                // listen to incoming messages from other publishers, forward them to the shim
                m_subscriber.ReceiveReady += OnSubscriberReady;

                // configure the beacon to listen on the broadcast port
                Console.WriteLine("Beacon is being configured to UDP port {0}", m_broadcastPort);
                m_beacon.Configure(m_broadcastPort);

                // publishing the random port to all other nodes
                Console.WriteLine("Beacon is publishing the Bus subscriber port {0}", m_randomPort);
                m_beacon.Publish(m_randomPort.ToString(), TimeSpan.FromSeconds(1));

                // Subscribe to all beacon on the port
                Console.WriteLine("Beacon is subscribing to all beacons on UDP port {0}", m_broadcastPort);
                m_beacon.Subscribe("");

                // listen to incoming beacons
                m_beacon.ReceiveReady += OnBeaconReady;

                // Create a timer to clear dead nodes
                NetMQTimer timer = new NetMQTimer(TimeSpan.FromSeconds(1));
                timer.Elapsed += ClearDeadNodes;

                // Create and configure the poller with all sockets and the timer
                m_poller = new NetMQPoller { m_shim, m_subscriber, m_beacon, timer };

                // signal the actor that we finished with configuration and
                // ready to work
                m_shim.SignalOK();

                // polling until cancelled
                m_poller.Run();
            }
        }

        private void OnShimReady(object sender, NetMQSocketEventArgs e)
        {
            // new actor command
            string command = m_shim.ReceiveFrameString();

            // check if we received end shim command
            if (command == NetMQActor.EndShimMessage)
            {
                // we cancel the socket which dispose and exist the shim
                m_poller.Stop();
            }
            else if (command == PublishCommand)
            {
                // it is a publish command
                // we just forward everything to the publisher until end of message
                NetMQMessage message = m_shim.ReceiveMultipartMessage();
                m_publisher.SendMultipartMessage(message);
            }
            else if (command == GetHostAddressCommand)
            {
                var address = m_beacon.BoundTo + ":" + m_randomPort;
                m_shim.SendFrame(address);
            }
        }

        private void OnSubscriberReady(object sender, NetMQSocketEventArgs e)
        {
            // we got a new message from the bus
            // let's forward everything to the shim
            NetMQMessage message = m_subscriber.ReceiveMultipartMessage();
            m_shim.SendMultipartMessage(message);
        }

        private void OnBeaconReady(object sender, NetMQBeaconEventArgs e)
        {
            // we got another beacon
            // let's check if we already know about the beacon
            var message = m_beacon.Receive();
            int port;
            int.TryParse(message.String, out port);

            NodeKey node = new NodeKey(message.PeerHost, port);

            // check if node already exist
            if (!m_nodes.ContainsKey(node))
            {
                // we have a new node, let's add it and connect to subscriber
                m_nodes.Add(node, DateTime.Now);
                m_publisher.Connect(node.Address);
                m_shim.SendMoreFrame(AddedNodeCommand).SendFrame(node.Address);
            }
            else
            {
                //Console.WriteLine("Node {0} is not a new beacon.", node);
                m_nodes[node] = DateTime.Now;
            }
        }

        private void ClearDeadNodes(object sender, NetMQTimerEventArgs e)
        {
            // create an array with the dead nodes
            var deadNodes = m_nodes.
                Where(n => DateTime.Now > n.Value + m_deadNodeTimeout)
                .Select(n => n.Key).ToArray();

            // remove all the dead nodes from the nodes list and disconnect from the publisher
            foreach (var node in deadNodes)
            {
                m_nodes.Remove(node);
                m_publisher.Disconnect(node.Address);
                m_shim.SendMoreFrame(RemovedNodeCommand).SendFrame(node.Address);
            }
        }
    }
}